// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: xguo

#include "coin2/exchange/base/order/order_context_manager.h"

#include <cfloat>
#include <cmath>
#include <iomanip>
#include <tuple>
#include <utility>

#include <fmt/core.h>

#include "coin/proto/coin_order_enums.pb.h"
#include "coin/proto/coin_order_gateway.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/exchange/base/order/system_critical_error.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "naviasync/appinfo/build_info.h"

namespace coin2::exchange::base::order {

namespace {

using coin::proto::OrderEvent;
using OrderSide = coin::proto::OrderSide;
using TradeSide = coin::proto::TradeSide;

bool IsBuy(OrderSide order_side) {
  return (
      order_side == OrderSide::BUY_ORDER || order_side == OrderSide::BUY_OPEN_ORDER ||
      order_side == OrderSide::BUY_CLOSE_ORDER);
}

bool IsSell(OrderSide order_side) {
  return (
      order_side == OrderSide::SELL_ORDER || order_side == OrderSide::SELL_OPEN_ORDER ||
      order_side == OrderSide::SELL_CLOSE_ORDER);
}

TradeSide ToTradeSide(OrderSide order_side) {
  if (IsBuy(order_side)) {
    return TradeSide::TRADE_BUY_SIDE;
  } else if (IsSell(order_side)) {
    return TradeSide::TRADE_SELL_SIDE;
  } else {
    NOTREACHED() << OrderSide_Name(order_side);
  }
  return TradeSide::TRADE_NO_SIDE;
}

auto GetDoubleEpsilonBySpecifyNumber(double x) -> double {
  return std::max((std::nextafter(x, DBL_MAX) - x) * 100, 1e-9);
}

}  // namespace

bool FillManager::IsReasonableFilledQty(double filled_qty, double orig_qty) {
  constexpr double ratio_upper_bound = 1.1;
  if (filled_qty - orig_qty < GetDoubleEpsilonBySpecifyNumber(orig_qty)) {
    return true;
  }
  if (filled_qty / orig_qty < ratio_upper_bound) {
    // Maybe formatting issue, still accept
    LOG(WARNING) << fmt::format(
        "Maybe formatting issue!! filled_qty: {:f}, orig_qty: {:f}",
        filled_qty,
        orig_qty);
    return true;
  }

  return false;
}

bool FillManager::IsFullyFilled() const {
  if (qty_accum_fill_ > qty_orig_) {
    return true;
  }
  return qty_orig_ - qty_accum_fill_ < GetDoubleEpsilonBySpecifyNumber(qty_orig_);
}

bool OrderContextUpdateResult::has_debug_info() const {
  // check empty
  return !extra_info.empty();
}

void OrderContextUpdateResult::Clear() {
  order_event.Clear();
  extra_info.clear();
}

void OrderContextUpdateResult::AddInfo(const std::string& location, const nlohmann::json& info) {
  extra_info[location].push_back(info);
  CreateDebugOrderEvent();
}

void OrderContextUpdateResult::CreateDebugOrderEvent() {
  auto build_info = GetBuildInfo();

  extra_info["og_version"] = build_info.revision();
  order_event.set_type(OrderEvent::DEBUG_INFO);
  order_event.set_tag(extra_info.dump());
}

bool OrderContext::IsClosible() const {
  return working_qty() < GetDoubleEpsilonBySpecifyNumber(order_spec_.qty);
}

void OrderContext::HandleOrderSubmitted(int64_t event_time) {
  submit_manager_.HandleSubmitted(event_time);
  UpdateLastSeenTime(event_time);
  // SUBMITTED,BEGIN,BEGIN
  CHECK_EQ(submit_manager_.status(), SubmitStatus::SUBMITTED);
  CHECK_EQ(cancel_manager_.status(), CancelStatus::BEGIN);
  CHECK_EQ(fill_manager_.status(), FillStatus::BEGIN);

  CHECK_EQ(status_, OrderSimpleStatus::CREATED_ORDER);
  status_ = OrderSimpleStatus::PENDING_ORDER;

  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::ORDER_SUBMITTED);
  oe.set_order_created_time(event_time);
  oe.set_order_trigger_time(event_time);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleOrderAccepted(int64_t event_time, const std::string& exchange_order_id) {
  submit_manager_.HandleAccepted(event_time);
  UpdateLastSeenTime(event_time);
  // ACCEPTED,BEGIN,(BEGIN, PART_FILLED, FULLY_FILLED)
  CHECK_EQ(submit_manager_.status(), SubmitStatus::ACCEPTED);

  if (exchange_order_id_.empty()) {
    SetExchangeOrderId(exchange_order_id);
  } else {
    CHECK_EQ(exchange_order_id_, exchange_order_id)
        << exchange_order_id_ << "," << exchange_order_id;
  }

  // TODO(xguo): deal with HandleUnknownOrder
  switch (fill_manager_.status()) {
    case FillStatus::BEGIN: {
      // yuxuan: caused crash.
      // CHECK_EQ(status_, OrderSimpleStatus::PENDING_ORDER);
      status_ = OrderSimpleStatus::WORKING_ORDER;
      auto oe = CreateOrderEvent();
      // TODO(daniel): update amend order flag
      oe.set_type(OrderEvent::ORDER_ACCEPTED);
      PublishOrderEvent(std::move(oe));
      break;
    }
    case FillStatus::PART_FILLED: {
      break;
    }
    case FillStatus::FULLY_FILLED: {
      // Order accepted message comes later then fill message.
      CHECK_EQ(status_, OrderSimpleStatus::DEAD_ORDER);
      break;
    }
    default: {
      NOTREACHED();
    }
  }
  CHECK(is_accepted());
}

void OrderContext::HandleOrderRejected(
    int64_t event_time,
    coin::proto::OrderErrorCode error_code,
    const std::string& exchange_error_code,
    const std::string& exchange_error_msg) {
  submit_manager_.HandleRejected(event_time);
  UpdateLastSeenTime(event_time);
  // REJECTED,BEGIN,BEGIN
  CHECK_EQ(submit_manager_.status(), SubmitStatus::REJECTED);
  CHECK_EQ(fill_manager_.status(), FillStatus::BEGIN);

  CHECK_EQ(status_, OrderSimpleStatus::PENDING_ORDER);
  status_ = OrderSimpleStatus::DEAD_ORDER;
  auto oe = CreateOrderEvent();
  // TODO(daniel): update amend order flag
  oe.set_type(OrderEvent::ORDER_REJECTED);
  oe.set_error_code(error_code);
  oe.set_exchange_error_code(exchange_error_code);
  oe.set_exchange_error_msg(exchange_error_msg);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleOrderError(int64_t event_time) {
  submit_manager_.HandleError(event_time);
  UpdateLastSeenTime(event_time);
  // ERROR,BEGIN,BEGIN
  CHECK_EQ(submit_manager_.status(), SubmitStatus::ERROR);
  CHECK_EQ(fill_manager_.status(), FillStatus::BEGIN);

  CHECK_EQ(status_, OrderSimpleStatus::PENDING_ORDER);
  status_ = OrderSimpleStatus::DEAD_ORDER;
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::ORDER_ERROR);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleCancelSubmitted(int64_t event_time) {
  cancel_manager_.HandleSubmitted(event_time);
  UpdateLastSeenTime(event_time);
  // ACCEPTED SUBMITTED (PART_FILLED FULLY_FILLED)
  CHECK(
      submit_manager_.status() == SubmitStatus::ACCEPTED ||
      submit_manager_.status() == SubmitStatus::SUBMITTED)
      << submit_manager_.status();
  // jhkim: likely cause crash
  // CHECK_EQ(cancel_manager_.status(), CancelStatus::SUBMITTED);
  CHECK(
      fill_manager_.status() == FillStatus::BEGIN ||
      fill_manager_.status() == FillStatus::PART_FILLED)
      << static_cast<int>(status_);
  CHECK(status_ == OrderSimpleStatus::PENDING_ORDER || status_ == OrderSimpleStatus::WORKING_ORDER)
      << static_cast<int>(status_);
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::CANCEL_SUBMITTED);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleCancelAccepted(int64_t event_time) {
  cancel_manager_.HandleAccepted(event_time);
  UpdateLastSeenTime(event_time);
  // ACCEPTED ACCEPTED (any fill status)
  // yuxuan: caused crash.
  // CHECK_EQ(submit_manager_.status(), SubmitStatus::ACCEPTED);
  // jhkim: caused crash.
  // CHECK_EQ(cancel_manager_.status(), CancelStatus::ACCEPTED);
  // yuxuan: caused crash.
  // CHECK_EQ(status_, OrderSimpleStatus::WORKING_ORDER);
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::CANCEL_ACCEPTED);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleCancelRejected(
    int64_t event_time,
    coin::proto::OrderErrorCode error_code,
    const std::string& exchange_error_code,
    const std::string& exchange_error_msg) {
  cancel_manager_.HandleRejected(event_time);
  UpdateLastSeenTime(event_time);
  // ACCEPTED REJECTED (any fill status)
  CHECK_EQ(cancel_manager_.last_status(), CancelStatus::REJECTED);
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::CANCEL_REJECTED);
  oe.set_error_code(error_code);
  oe.set_exchange_error_code(exchange_error_code);
  oe.set_exchange_error_msg(exchange_error_msg);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleCancelError(int64_t event_time) {
  cancel_manager_.HandleError(event_time);
  UpdateLastSeenTime(event_time);
  // ACCEPTED ACCEPTED (any fill status)
  // CHECK_EQ(submit_manager_.status(), SubmitStatus::ACCEPTED);
  CHECK_EQ(cancel_manager_.last_status(), CancelStatus::ERROR);
  // CHECK_EQ(status_, OrderSimpleStatus::WORKING_ORDER);
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::CANCEL_ERROR);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleCancelConfirmed(int64_t event_time, double canceled_qty) {
  cancel_manager_.HandleConfirmed(event_time, canceled_qty);
  UpdateLastSeenTime(event_time);
  // ACCEPTED Confirmed (any fill status)
  // CHECK_EQ(submit_manager_.status(), SubmitStatus::ACCEPTED);
  CHECK_EQ(cancel_manager_.status(), CancelStatus::CONFIRMED);
  if (IsClosible()) {
    status_ = OrderSimpleStatus::DEAD_ORDER;
  }
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::CANCEL_CONFIRMED);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleCancelConfirmed(int64_t event_time) {
  HandleCancelConfirmed(event_time, working_qty());
}

void OrderContext::HandleOrderAutoCanceled(int64_t event_time, double canceled_qty) {
  cancel_manager_.HandleAutoCanceled(event_time, canceled_qty);
  UpdateLastSeenTime(event_time);
  // ACCEPTED Confirmed (any fill status)
  CHECK(
      submit_manager_.status() == SubmitStatus::SUBMITTED ||
      submit_manager_.status() == SubmitStatus::ACCEPTED ||
      submit_manager_.status() == SubmitStatus::ERROR);
  if (IsClosible()) {
    status_ = OrderSimpleStatus::DEAD_ORDER;
  }
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::ORDER_AUTO_CANCELED);
  PublishOrderEvent(std::move(oe));
}

void OrderContext::HandleFill(
    int64_t event_time,
    double price_last_fill,
    double qty_last_fill,
    const std::string& fill_id,
    coin::proto::FillType fill_type,
    coin::proto::FillSource fill_source,
    std::optional<double> fee,
    std::optional<std::string> fee_currency) {
  UpdateLastSeenTime(event_time);

  // When order state is known to be fully_filled, but fill detail is unknown.
  // We publish inferred fill to strategy only.
  if (fill_source == coin::proto::FillSource::IMPLIED_FROM_DEAD) {
    auto oe = CreateOrderEvent();
    oe.set_type(OrderEvent::ORDER_FILLED);
    oe.set_fill_price(price_last_fill);
    oe.set_fill_qty(qty_last_fill);
    oe.set_fill_id(fill_id);
    oe.set_fill_type(fill_type);
    oe.set_fully_filled(true);
    oe.set_publish_live(true);
    oe.set_fill_source(fill_source);
    if (fee) {
      oe.set_fee(*fee);
    }
    if (fee_currency) {
      oe.set_fee_currency(*fee_currency);
    }
    implied_fill_published_ = true;
    PublishOrderEvent(std::move(oe));
    return;
  }

  if (!fill_manager_.HasFill(fill_id)) {
    bool success = fill_manager_.UpdateByLastFill(
        order_spec_.order_side,
        price_last_fill,
        qty_last_fill,
        fill_id,
        fill_type,
        last_seen_time_);
    if (success && replaced_proc_order_id_ > 0) {
      auto* replaced_oc = ocm_->FindByProcOrderId(replaced_proc_order_id_);
      if (replaced_oc) {
        const auto* fill = fill_manager_.last_fill();
        replaced_oc->UpdateByLastFill(
            fill->timestamp,
            fill->side,
            fill->price,
            fill->qty,
            fill->fill_id,
            fill->fill_type);
      }
    }
    if (IsClosible()) {
      status_ = OrderSimpleStatus::DEAD_ORDER;
    }
    if (success) {
      auto oe = CreateOrderEvent();
      oe.set_type(OrderEvent::ORDER_FILLED);
      oe.set_fill_price(price_last_fill);
      oe.set_fill_qty(qty_last_fill);
      oe.set_fill_id(fill_id);
      oe.set_fill_type(fill_type);
      oe.set_fully_filled(fill_manager().IsFullyFilled());
      oe.set_publish_live(!implied_fill_published_);
      oe.set_duplicate(implied_fill_published_);
      if (fee) {
        oe.set_fee(*fee);
      }
      if (fee_currency) {
        oe.set_fee_currency(*fee_currency);
      }
      PublishOrderEvent(std::move(oe));
    }
  } else {
    auto* fill_element = fill_manager_.GetMutableFill(fill_id);
    CHECK(fill_element) << "can't reach here.";
    if (fill_element->fill_type == coin::proto::FillType::UNKNOWN_FILL_TYPE &&
        fill_type != coin::proto::FillType::UNKNOWN_FILL_TYPE) {
      // Handle Okex case, usually fill type comes in later rest fill msg.
      fill_element->fill_type = fill_type;
      auto oe = CreateOrderEvent();
      oe.set_type(OrderEvent::ORDER_FILLED);
      oe.set_fill_price(price_last_fill);
      oe.set_fill_qty(qty_last_fill);
      oe.set_fill_id(fill_id);
      oe.set_fill_type(fill_type);
      oe.set_fully_filled(fill_manager().IsFullyFilled());
      oe.set_publish_live(false);  // Do not publish to strategy
      oe.set_duplicate(true);
      if (fee) {
        oe.set_fee(*fee);
      }
      if (fee_currency) {
        oe.set_fee_currency(*fee_currency);
      }
      PublishOrderEvent(std::move(oe));
    }
  }
}

void OrderContext::ReplacedByOrder(int64_t timestamp, OrderContext* oc) {
  if (cancel_manager().status() == CancelStatus::BEGIN ||
      cancel_manager().status() == CancelStatus::REJECTED) {
    HandleOrderAutoCanceled(timestamp, working_qty());
  } else {
    if (!IsCancelAccepted()) {
      HandleCancelAccepted(timestamp);
    }
    HandleCancelConfirmed(timestamp, working_qty());
  }
  // transfer fill to new oc
  oc->SetAmendedCnt(GetAmendedCnt() + 1);
  auto fills = fill_manager_.GetFills();
  std::sort(fills.begin(), fills.end(), [](const FillElement* lhs, const FillElement* rhs) -> bool {
    return lhs->timestamp < rhs->timestamp;
  });
  for (const auto* fill : fills) {
    oc->UpdateByLastFill(
        fill->timestamp,
        fill->side,
        fill->price,
        fill->qty,
        fill->fill_id,
        fill->fill_type);
  }
  replaced_proc_order_id_ = oc->proc_order_id();
  // erase self from live orders
  ocm_->SetDeadWithErrorCode(proc_order_id(), coin::proto::OrderErrorCode::ORDER_REPLACED);
}

void OrderContext::UpdateByLastFill(
    int64_t event_time,
    coin::proto::OrderSide order_side,
    double price_last_fill,
    double qty_last_fill,
    const std::string& fill_id,
    coin::proto::FillType fill_type) {
  if (fill_manager_.HasFill(fill_id)) {
    return;
  }
  fill_manager_
      .UpdateByLastFill(order_side, price_last_fill, qty_last_fill, fill_id, fill_type, event_time);
}

void OrderContext::HandleProcIdReplaced(int64_t event_time, const OrderContext& oc_alt) {
  UpdateLastSeenTime(event_time);
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::PROC_ID_REPLACED);
  PublishOrderEvent(std::move(oe));
  oc_alt.order_spec().mutable_tag = order_spec_.tag;
}

void OrderContext::HandleUnknownOrder(int64_t event_time) {
  UpdateLastSeenTime(event_time);
  CHECK(has_exchange_order_id());

  is_foreign_ = true;
  status_ = OrderSimpleStatus::CREATED_ORDER;
  HandleOrderSubmitted(event_time);
  HandleOrderAccepted(event_time, exchange_order_id_);
  CHECK_EQ(status_, OrderSimpleStatus::WORKING_ORDER);
  auto oe = CreateOrderEvent();
  oe.set_type(OrderEvent::UNKNOWN_ORDER_CREATED);
  PublishOrderEvent(std::move(oe));
}

coin::proto::OrderEvent OrderContext::CreateOrderEvent() {
  coin::proto::OrderEvent oe;
  if (has_exchange_order_id()) {
    oe.set_external_order_id(exchange_order_id_);
  }

  oe.set_event_time(last_seen_time_);
  oe.set_order_state(status_);
  oe.set_proc_order_id(proc_order_id_);
  oe.set_exchange_type(order_spec_.exchange);
  oe.set_market_type(order_spec_.market_type);
  oe.set_native_symbol(order_spec_.product);
  oe.set_symbol(order_spec_.norm_product);
  oe.set_order_price(order_spec_.price);
  oe.set_order_qty(order_spec_.qty);
  oe.set_order_side(order_spec_.order_side);
  oe.set_trade_side(ToTradeSide(order_spec_.order_side));
  oe.set_order_type(order_spec_.order_type);
  oe.set_tag(order_spec_.mutable_tag.empty() ? order_spec_.tag : order_spec_.mutable_tag);
  oe.set_duration(order_spec_.duration);
  oe.set_post_only(order_spec_.post_only);
  oe.set_cancelled_qty(cancel_manager_.canceled_qty());

  if (order_spec_.use_amend) {
    oe.set_use_amend(order_spec_.use_amend);
    oe.set_orig_proc_order_id(order_spec_.orig_proc_order_id);
  }

  return oe;
}

void OrderContext::PublishOrderEvent(coin::proto::OrderEvent&& oe) {
  if (nullptr == ocm_) {
    LOG(ERROR) << "[OrderContext] ocm is not set";
    return;
  }
  ocm_->PublishOrderEvent(std::move(oe), false);
}

#define LOCATION std::string(__func__) + ":" + std::to_string(__LINE__)

namespace {

using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using SubmitOrderRersponse = coin::proto::SubmitOrderResponse;
using coin::proto::CancelOrderResponse;
using coin::proto::OrderDirection;
using coin::proto::ProductOrderElement;

std::string StringTableToString(const std::vector<std::vector<std::string>>& table_str) {
  std::vector<int> max_length_per_column;
  int max_column = 0;
  int row_size = table_str.size();
  for (int i = 0; i < row_size; i++) {
    int column_size = table_str[i].size();
    if (column_size > max_column) {
      max_column = column_size;
    }
    for (int j = 0; j < column_size; j++) {
      if (max_length_per_column.size() < j + 1) {
        max_length_per_column.push_back(table_str[i][j].size());
      } else if (max_length_per_column[j] < table_str[i][j].size()) {
        max_length_per_column[j] = table_str[i][j].size();
      }
    }
  }

  std::stringstream output_str;
  output_str << "\n";
  for (int i = 0; i < row_size; i++) {
    int column_size = table_str[i].size();
    for (int j = 0; j < column_size; j++) {
      output_str << std::setw(max_length_per_column[j] + 2) << table_str[i][j];
    }
    output_str << "\n";
    if (i == 0) {
      for (int j = 0; j < max_column; j++) {
        output_str << " " << std::string(max_length_per_column[j] + 1, '-');
      }
      output_str << "\n";
    }
  }
  return output_str.str();
}

}  // namespace

void OrderContextManager::OrderEventPublisher::PublishOrderEvent(
    coin::proto::OrderEvent&& oe,
    bool flush) {
  oe_to_publish_pending_.emplace_back(std::move(oe));
  if (flush) {
    Flush();
  }
}

void OrderContextManager::OrderEventPublisher::Reset() {
  if (!oe_to_publish_pending_.empty() || !oe_to_publish_ongoing_.empty()) {
    LOG_EVERY_N(WARNING, 20)
        << "@og team, order event publishing is pending. re-entry happens. please check";
  }
  // TODO(daniel): let's not clear first
  // oe_to_publish_pending_.clear();
  // oe_to_publish_ongoing_.clear();
}

void OrderContextManager::OrderEventPublisher::Flush() {
  if (oe_to_publish_pending_.empty()) {
    return;
  }
  if (!handler_list_) {
    // LOG(ERROR) << "@og team, order event is abandoned due to handler is not set. please check";
    oe_to_publish_pending_.clear();  // just throw oe if handle_list_ is not set
    return;
  }

  bool unprogress = false;
  bool inprogress = true;
  if (!publish_in_progress_.compare_exchange_strong(unprogress, inprogress)) {
    LOG_EVERY_N(WARNING, 20)
        << "@og team, order event publishing is ongoing. re-entry happens. please check";
    return;
  }
  int debug_flush_round_cnt = 0;
  while (!oe_to_publish_pending_.empty()) {  // TODO(daniel): is this too danger ?
    if (++debug_flush_round_cnt > 2) {
      LOG(WARNING) << "@og team, please check if it's due to function re-entried too many times: "
                   << debug_flush_round_cnt;
    }
    oe_to_publish_ongoing_.swap(oe_to_publish_pending_);
    for (auto& oe : oe_to_publish_ongoing_) {
      for (auto& handler : (*handler_list_)) {
        handler(oe);
      }
    }
    oe_to_publish_ongoing_.clear();
  }
  publish_in_progress_.compare_exchange_strong(inprogress, unprogress);
}

void OrderContextManager::PublishOrderEvent(coin::proto::OrderEvent&& oe, bool flush) {
  if (nullptr == order_event_publisher_) {
    LOG(ERROR) << "[OrderContext] order event publisher is not set";
    return;
  }
  order_event_publisher_->PublishOrderEvent(std::move(oe), flush);
}

void OrderContextManager::Reset() {
  if (nullptr == order_event_publisher_) {
    LOG(ERROR) << "[OrderContext] order event publisher is not set";
    return;
  }
  order_event_publisher_->Reset();
}

void OrderContextManager::Flush() {
  if (nullptr == order_event_publisher_) {
    LOG(ERROR) << "[OrderContext] order event publisher is not set";
    return;
  }
  order_event_publisher_->Flush();
}

bool OrderContextManager::HasProcOrderId(int64_t proc_order_id) const {
  return proc_order_id_map_.find(proc_order_id) != std::end(proc_order_id_map_);
}

bool OrderContextManager::HasExchangeOrderId(const std::string& exchange_order_id) const {
  if (exchange_order_id.empty()) {
    return false;
  }
  return exchange_order_id_map_.find(exchange_order_id) != std::end(exchange_order_id_map_);
}

bool OrderContextManager::HasOrder(const OrderContext& order_context) const {
  return (
      HasProcOrderId(order_context.proc_order_id()) ||
      HasExchangeOrderId(order_context.exchange_order_id()));
}

auto OrderContextManager::FindByProcOrderId(const int64_t proc_order_id) -> OrderContext* {
  if (HasProcOrderId(proc_order_id)) {
    return proc_order_id_map_.at(proc_order_id);
  } else {
    return nullptr;
  }
}

auto OrderContextManager::FindByProcOrderId(const int64_t proc_order_id) const
    -> const OrderContext* {
  if (HasProcOrderId(proc_order_id)) {
    return proc_order_id_map_.at(proc_order_id);
  } else {
    return nullptr;
  }
}

auto OrderContextManager::FindByExchangeOrderId(const std::string& exchange_order_id)
    -> OrderContext* {
  if (HasExchangeOrderId(exchange_order_id)) {
    return exchange_order_id_map_.at(exchange_order_id);
  } else {
    return nullptr;
  }
}

auto OrderContextManager::FindSubmitPending(const coin::proto::ProductOrderElement& order_info)
    -> OrderContext* {
  const static int64_t kDeltaTs = 1'000'000'000LL;
  const static double kDelta = 1e-9;

  std::vector<const OrderContext*> candid;
  auto oc_list = GetAllProductWorkingOrders(order_info.symbol());
  for (const auto& oc : oc_list) {
    const auto& order_spec = oc->order_spec();
    if (!oc->has_exchange_order_id() && order_spec.order_side == order_info.side() &&
        std::abs(order_spec.price - order_info.price()) < kDelta &&
        std::abs(order_spec.qty - order_info.qty()) < kDelta &&
        std::abs(order_spec.order_created_time - order_info.created_timestamp()) < kDeltaTs) {
      candid.push_back(oc);
    }
  }
  // 0: not found
  // more than 1: ambiguious
  if (candid.size() == 1) {
    return const_cast<OrderContext*>(candid.back());
  } else {
    return nullptr;
  }
}

auto OrderContextManager::FindByOrderIds(
    const std::string& symbol,
    const std::string& exchange_order_id,
    int64_t proc_order_id) -> OrderContext* {
  // TODO(daniel): check some scenarios
  // - duplicated proc_order_id by different symbol
  // - duplicated exchange_order_id by different symbol
  // - FindByExchangeOrderId for multi order contexts share same exchange order id, e.g amend order
  OrderContext* oc = nullptr;
  // theoretically, proc_order_id as the unique key to identify an order context
  if (0 != proc_order_id) {
    oc = FindByProcOrderId(proc_order_id);
    if (oc && oc->order_spec().product != symbol) {
      oc = nullptr;
    }
  }
  // for some cases, it may need to find order context by exchange_order_id
  // - for exchange which does not attach client order id in order
  // - for amend order in different process
  if (nullptr == oc) {
    oc = FindByExchangeOrderId(exchange_order_id);
    if (oc && 0 != proc_order_id && oc->proc_order_id() != proc_order_id) {
      oc = IterateSearchByOrderIds(exchange_order_id, proc_order_id);
    }
  }
  return oc;
}

auto OrderContextManager::IterateSearchByOrderIds(
    const std::string& exchange_order_id,
    int64_t proc_order_id) -> OrderContext* {
  const auto key = std::make_pair(exchange_order_id, proc_order_id);
  auto iter = multi_orders_map_.find(key);
  if (iter != multi_orders_map_.end()) {
    return iter->second;
  }
  return nullptr;
}

void OrderContextManager::RegisterOrderEventHandler(
    const OrderContext::OrderEventHandler& handler) {
  handler_list_.push_back(handler);
  if (order_event_publisher_) {
    order_event_publisher_->RegisterOrderEventHandlers(&handler_list_);
  }
}
void OrderContextManager::RegisterRiskEventHandler(const RiskEventHandler& handler) {
  if (!risk_pending_order_) {
    return;
  }
  risk_pending_order_->RegisterRiskEventHandler(handler);
}

auto OrderContextManager::GetLiveOrderSummary() const -> OrderContextManagerSummaryProto {
  OrderContextManagerSummaryProto order_summary_list;
  for (const auto& pair : live_orders_) {
    auto* const oc = pair.second;
    auto* summary = order_summary_list.add_summary();
    const auto order_spec = oc->order_spec();
    summary->set_symbol(order_spec.product);
    summary->set_proc_order_id(oc->proc_order_id());
    if (oc->has_exchange_order_id()) {
      summary->set_exchange_order_id(oc->exchange_order_id());
    }
    summary->set_order_side(order_spec.order_side);
    summary->set_order_type(order_spec.order_type);
    summary->set_order_price(order_spec.price);
    summary->set_order_qty(order_spec.qty);
    summary->set_order_submitted_time(oc->submit_manager().submitted_time());
    summary->set_cancel_submitted_time(oc->cancel_created_time());
  }
  return order_summary_list;
}

void OrderContextManager::RemoveTimeoutPendingOrders(int64_t timestamp) {
  std::vector<int64_t> proc_order_ids;
  for (auto& pair : live_orders_) {
    auto* const oc = pair.second;
    const auto order_spec = oc->order_spec();
    const int64_t age = timestamp - order_spec.order_created_time;
    if (age > submit_pending_timeout_ && !oc->has_exchange_order_id()) {
      proc_order_ids.push_back(oc->proc_order_id());
      risk_pending_order_->HandleOrderTimedout(timestamp, *oc);
    }
  }

  for (int64_t proc_order_id : proc_order_ids) {
    SetDead(proc_order_id);
    SPDLOG_WARN("Pending submit order removed! {}", proc_order_id);
  }

  if (!proc_order_ids.empty()) {
    nlohmann::json json;
    json["submit_pending_timeout"] = submit_pending_timeout_;
    json["proc_order_id"] = proc_order_ids;
    json["error_code"] = __func__;
    result_.AddInfo(LOCATION, json);
  }
}

void OrderContextManager::ManageRisk(int64_t timestamp) {
  if (!risk_pending_order_) {
    return;
  }
  risk_pending_order_->ManageRisk(timestamp);
}

std::vector<int64_t> OrderContextManager::GetZombieOrders(int64_t timestamp, int64_t max_age) {
  std::vector<int64_t> proc_order_ids;
  for (auto& pair : live_orders_) {
    auto* const oc = pair.second;
    const int64_t age = (timestamp - oc->last_time_query_alive());
    if (age > max_age && oc->last_time_query_alive() > 0) {
      proc_order_ids.push_back(oc->proc_order_id());
    }
  }
  return proc_order_ids;
}

void OrderContextManager::RemoveZombieOrders(int64_t timestamp, int64_t max_order_idle_age) {
  auto proc_order_ids = GetZombieOrders(timestamp, max_order_idle_age);
  for (int64_t proc_order_id : proc_order_ids) {
    SetDead(proc_order_id);
    SPDLOG_ERROR("Zombie order removed! {}", proc_order_id);
  }

  if (!proc_order_ids.empty()) {
    nlohmann::json json;
    json["proc_order_id"] = proc_order_ids;
    json["max_query_alive_time"] = max_order_idle_age;
    json["error_code"] = __func__;
    result_.AddInfo(LOCATION, json);
  }
}

std::vector<int64_t> OrderContextManager::GetCancelRejectedOrders() {
  std::vector<int64_t> proc_order_ids;
  for (auto& pair : live_orders_) {
    auto* const oc = pair.second;
    if (oc->cancel_manager().num_of_cancels() > 2) {
      proc_order_ids.push_back(oc->proc_order_id());
    }
  }
  return proc_order_ids;
}

void OrderContextManager::SetTimeoutCancelRejected(
    const OrderContext* order_context,
    int64_t timestamp) {
  coin::proto::CancelOrderResponse resp;

  resp.set_symbol(order_context->order_spec().product);
  resp.set_proc_order_id(order_context->proc_order_id());
  resp.set_http_status(200);
  resp.set_exchange_order_id(order_context->exchange_order_id());
  resp.set_timestamp(timestamp);
  resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
  resp.set_success(false);
  HandleCancelResponse(resp);

  // abnormal: cancel timeout
  nlohmann::json json;
  json["proc_order_id"] = order_context->proc_order_id();
  json["error_code"] = __func__;
  result_.AddInfo(LOCATION, json);
}

std::string OrderContextManager::DebugString(std::optional<int64_t> timestamp) const {
  std::vector<std::vector<std::string>> table_str;
  std::stringstream ss;
  table_str.push_back(
      {"order_id",
       "exchange_order_id",
       "product",
       "price",
       "qty",
       "side",
       "is_foreign",
       "cancel_sent",
       "age"});
  for (auto pair : live_orders_) {
    std::vector<std::string> str_row;
    auto* const oc = pair.second;
    const auto order_spec = oc->order_spec();
    str_row.push_back(std::to_string(oc->proc_order_id()));
    if (oc->has_exchange_order_id()) {
      str_row.push_back(oc->exchange_order_id());
    } else {
      str_row.emplace_back("");
    }
    str_row.push_back(order_spec.product);
    ss.str("");
    ss << std::fixed << std::setprecision(8) << order_spec.price;
    str_row.push_back(ss.str());
    ss.str("");
    ss << std::fixed << std::setprecision(8) << order_spec.qty;
    str_row.push_back(ss.str());
    str_row.push_back(coin::proto::OrderSide_Name(order_spec.order_side));
    if (oc->IsForeign()) {
      str_row.push_back("True");
    } else {
      str_row.push_back("False");
    }
    if (oc->cancel_created_time() > 0) {
      str_row.push_back("True");
    } else {
      str_row.push_back("False");
    }
    auto ts = oc->order_spec().order_created_time;
    if (ts == 0) {
      ts = oc->submit_manager().submitted_time();
    }
    int age = (timestamp.value_or(current_ts_) - ts) / 1e9;
    str_row.push_back(std::to_string(age));
    table_str.push_back(str_row);
  }
  return StringTableToString(table_str);
}

void OrderContextManager::SetDead(int64_t proc_order_id) {
  SetDeadWithErrorCode(proc_order_id, coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR);
}

void OrderContextManager::SetDeadWithErrorCode(
    int64_t proc_order_id,
    coin::proto::OrderErrorCode error_code) {
  SetDeadIterator(live_orders_.find(proc_order_id), error_code);
}

std::unordered_map<int64_t, OrderContext*>::iterator OrderContextManager::SetDeadIterator(
    std::unordered_map<int64_t, OrderContext*>::iterator it,
    coin::proto::OrderErrorCode error_code) {
  if (it != live_orders_.end()) {
    int64_t proc_order_id = it->first;
    OrderContext* oc = it->second;
    product_live_orders_.at(oc->order_spec().product).erase(proc_order_id);
    SPDLOG_INFO(
        "[SetDead] Set buy_fill, sell_fill potential -->> {} : {}",
        oc->order_spec().product,
        oc->filled_qty());
    if (oc->order_spec().direction() == OrderDirection::DIRECTION_BUY) {
      fill_not_synced_to_buy_pos_[oc->order_spec().product] += oc->filled_qty();
    } else if (oc->order_spec().direction() == OrderDirection::DIRECTION_SELL) {
      fill_not_synced_to_sell_pos_[oc->order_spec().product] += oc->filled_qty();
    }
    oc->dead_timestamp(GetCurrentTimestamp());
    oc->dead_reason_code(error_code);
    it = live_orders_.erase(it);
  }

  return it;
}

void OrderContextManager::AddBackWorkingOrders(OrderContext* oc) {
  live_orders_[oc->proc_order_id()] = oc;
  product_live_orders_[oc->order_spec().product][oc->proc_order_id()] = oc;
  SPDLOG_INFO("[AddBackWorkingOrders]: {}", oc->ToProto().DebugString());
}

void OrderContextManager::SetDeadOrderAliveRestartSec(const MarketExchangeApi& mea) {
  dead_order_alive_restart_sec_ = 1000;
  std::ifstream ifs("data/coin2/order/og_info.json");
  nlohmann::json jf = nlohmann::json::parse(ifs);
  if (jf.find(mea.String()) != jf.end()) {
    auto data = jf[mea.String()];
    if (data.find("dead_order_alive_restart_sec") != data.end()) {
      dead_order_alive_restart_sec_ = data["dead_order_alive_restart_sec"];
    }
  }
}

void OrderContextManager::SetBookFeedFillInterval(const MarketExchangeApi& mea) {
  book_feed_fill_interval_ = 1000 * 1000'000'000LL;
  std::ifstream ifs("data/coin2/order/og_info.json");
  nlohmann::json jf = nlohmann::json::parse(ifs);
  if (jf.find(mea.String()) != jf.end()) {
    auto data = jf[mea.String()];
    if (data.find("book_feed_fill_interval_sec") != data.end()) {
      book_feed_fill_interval_ =
          static_cast<int64_t>(data["book_feed_fill_interval_sec"].get<double>() * 1000'000'000LL);
    }
  }
}

bool OrderContextManager::IsTickAligned(const OrderSpec& order_spec) const {
  if (!symcache_) {
    // Skip checking.
    return true;
  }

  auto product =
      CreateProductFromNativeString(mea_, order_spec.product, order_spec.order_created_time);
  auto product_holder = symcache_->GetProductHolder(*product, order_spec.order_created_time);
  auto formatter = product_holder->product_info().order_formatter();

  double price = stod(formatter.FormatPrice(order_spec.price).data());
  double qty = stod(formatter.FormatQty(order_spec.qty).data());
  return IsClose(price, order_spec.price) && IsClose(qty, order_spec.qty);
}

OrderContextHolder::~OrderContextHolder() { ocm->DeleteOrderMap(*order_context); }

OrderContext* OrderContextManager::AddOrderContext(std::unique_ptr<OrderContext>&& order_context) {
  CHECK_GT(order_context->order_spec().order_created_time, 0);
  CHECK_EQ(HasOrder(*order_context), false) << order_context->ToProto().DebugString();

  int loop_times = 0;
  while (order_list_.full()) {
    CHECK(++loop_times <= order_list_.capacity())
        << "CIRCULAR QUEUE is not sparse enough. capacity of queue: " << order_list_.capacity()
        << ", num of live order: " << order_list_.size();
    if (!order_list_.front()) {
      order_list_.pop_front();
      continue;
    }
    auto* order_ptr = order_list_.front().release();
    order_list_.pop_front();
    if (IsInWorkingOrders(order_ptr->order_context.get())) {
      order_list_.push_back(std::unique_ptr<OrderContextHolder>(order_ptr));
    } else if (order_ptr->order_context->is_permanent()) {
      order_list_.push_back(std::unique_ptr<OrderContextHolder>(order_ptr));
    } else {
      delete order_ptr;
    }
  }

  order_list_.push_back(std::make_unique<OrderContextHolder>(std::move(order_context), this));
  auto* oc = order_list_.back()->order_context.get();

  proc_order_id_map_[oc->proc_order_id()] = oc;

  if (oc->has_exchange_order_id()) {
    SetExchangeOrderId(oc, oc->exchange_order_id());
  }

  if (oc->status() != OrderSimpleStatus::DEAD_ORDER) {
    live_orders_[oc->proc_order_id()] = oc;
    product_live_orders_[oc->order_spec().product][oc->proc_order_id()] = oc;
  }
  oc->RegisterOrderEventHandlers(&handler_list_);
  oc->RegisterOrderContextManager(this);

  return oc;
}

OrderContext* OrderContextManager::AddOrderContext(const OrderSpec& order_spec) {
  if (order_spec.proc_order_id == 0) {
    order_spec.proc_order_id = proc_order_id_;
    proc_order_id_++;
  } else if (order_spec.proc_order_id < proc_order_id_) {
    CHECK_EQ(0, proc_order_id_map_.count(order_spec.proc_order_id))
        << "duplicate proc order id " << order_spec.proc_order_id;
  } else if (order_spec.proc_order_id > proc_order_id_) {
    proc_order_id_ = order_spec.proc_order_id + 1;
  } else {
    CHECK_EQ(order_spec.proc_order_id, proc_order_id_);
    proc_order_id_++;
  }

  return AddOrderContext(std::make_unique<OrderContext>(order_spec.proc_order_id, order_spec));
}

OrderContext* OrderContextManager::AddOrderContext(const ProductOrderElement& order_proto) {
  CHECK(order_proto.has_exchange_order_id()) << order_proto.DebugString();

  auto orig_proc_order_id = order_proto.orig_proc_order_id();
  OrderContext* orig_oc = orig_proc_order_id == 0 ? nullptr : FindByProcOrderId(orig_proc_order_id);
  // clone from original order spec if found it
  if (orig_oc) {
    auto order_spec = orig_oc->order_spec().CloneAsAmendOrder(
        orig_oc->proc_order_id(),
        order_proto.price(),
        order_proto.qty());
    if (0 != order_proto.proc_order_id()) {
      order_spec.proc_order_id = order_proto.proc_order_id();
    }
    auto new_oc = AddOrderContext(order_spec);
    SetExchangeOrderId(new_oc, order_proto.exchange_order_id());

    return new_oc;
  }

  // create new order spec based on proto
  auto product = CreateProductFromNativeString(
      MarketExchangeApi{
          order_proto.market_type(),
          order_proto.exchange_type(),
          order_proto.api_version()},
      order_proto.symbol(),
      order_proto.timestamp());

  OrderSpec order_spec(
      *product,
      order_proto.order_type(),
      order_proto.side(),
      coin::proto::OrderDuration::GTC_ORDER,
      order_proto.price(),
      order_proto.qty(),
      order_proto.timestamp(),
      order_proto.post_only(),
      order_proto.leverage_rate(),
      "add_from_order_proto");
  if (0 != order_proto.proc_order_id()) {
    order_spec.proc_order_id = order_proto.proc_order_id();
  }
  if (0 != orig_proc_order_id) {  // to keep order flow consistance
    order_spec.SetAmend(true);
    order_spec.SetOrigProcOrderId(orig_proc_order_id);
  }
  auto new_oc = AddOrderContext(order_spec);
  // tag is always empty from parser, except for bithumb virtual order.
  if (order_proto.has_tag() && order_proto.tag() == "permanent") {
    new_oc->set_permanent();
  }
  SetExchangeOrderId(new_oc, order_proto.exchange_order_id());

  return new_oc;
}

bool OrderContextManager::PostValidateOrderContext(const OrderContext* order_context) const {
  if (!IsTickAligned(order_context->order_spec())) {
    SPDLOG_WARN("price/qty in order spec not tick-aligned!");
    return false;
  }

  return true;
}

void OrderContextManager::SetExchangeOrderId(
    OrderContext* oc,
    const std::string& exchange_order_id) {
  SYSTEM_CRITICAL_ERROR(
      oc->proc_order_id() > 0,
      fmt::format(
          "fatal error, please contact OG team. info: {} - {}",
          exchange_order_id,
          oc->proc_order_id()));
  SYSTEM_CRITICAL_ERROR(
      !exchange_order_id.empty(),
      fmt::format(
          "fatal error, please contact OG team. info: {} - {}",
          exchange_order_id,
          oc->proc_order_id()));

  // TODO(daniel):
  // - always keep the latest oc for exchange_order_id_map_ based on exchange transaction time
  oc->SetExchangeOrderId(exchange_order_id);
  exchange_order_id_map_[exchange_order_id] = oc;

  const auto key = std::make_pair(exchange_order_id, oc->proc_order_id());
  bool inserted = false;
  std::tie(std::ignore, inserted) = multi_orders_map_.emplace(key, oc);

  SYSTEM_CRITICAL_ERROR(
      inserted,
      fmt::format(
          "fatal error, please contact OG team. info: {} - {}",
          exchange_order_id,
          oc->proc_order_id()));
}

void OrderContextManager::ValidateConsistency() {
  for (auto id_oc : live_orders_) {
    int64_t proc_order_id = id_oc.first;
    auto* oc = id_oc.second;
    CHECK_EQ(proc_order_id, oc->proc_order_id());
  }

  for (auto id_oc : proc_order_id_map_) {
    int64_t proc_order_id = id_oc.first;
    auto* oc = id_oc.second;
    CHECK_EQ(proc_order_id, oc->proc_order_id());
  }

  for (const auto& id_oc : exchange_order_id_map_) {
    const auto& exchange_order_id = id_oc.first;
    auto* oc = id_oc.second;
    CHECK_EQ(exchange_order_id, oc->exchange_order_id());
  }
}

void OrderContextManager::HandleAmendOrderAccepted(int64_t timestamp, OrderContext* oc) {
  if (!oc->IsAmendOrder()) {
    return;
  }
  if (!oc->is_accepted()) {
    return;
  }
  // handle cancel accepted for original order of amend order
  auto* orig_oc = FindByProcOrderId(oc->order_spec().orig_proc_order_id);
  if (!orig_oc) {
    return;  // amend order for cross process
  }
  orig_oc->ReplacedByOrder(timestamp, oc);
}

void OrderContextManager::HandleOrderInfoList(const ProductOrderBundle& product_order_bundle) {
  auto ts = get_ts();
  auto guard = unknown_order_id_manager_.GetGuard();

  auto order_event_publish_helper = GenerateOrderEventFlusher();

  for (const auto& order_info : product_order_bundle.each_order()) {
    // Each order_info must have exchange_order_id, better to have proc_order_id as well.
    CHECK(order_info.has_exchange_order_id()) << order_info.DebugString();

    if (!IsManagedSymbol(order_info.symbol())) {
      if (guard.IsNeedLog(order_info.symbol(), order_info.exchange_order_id())) {
        nlohmann::json json;
        json["error_code"] = "NOT_MANAGED_SYMBOL";
        json["order_info"] = nlohmann::json::parse(ProtoToJsonString(order_info));
        result_.AddInfo(LOCATION, json);
      }
      continue;
    }

    OrderContext* oc = nullptr;

    // TODO(daniel): apply to all exchanges that find order context by proc order id first
    if (mea().exchange == ExchangeType::Prex) {
      oc = FindByOrderIds(
          order_info.symbol(),
          order_info.exchange_order_id(),
          order_info.proc_order_id());
    } else {
      // Must have exchange_order_id, proc_order_id is not required, but better to have.
      oc = FindByExchangeOrderId(order_info.exchange_order_id());
      if (oc == nullptr) {
        // receive this before submit response
        if (order_info.has_proc_order_id() && (order_info.proc_order_id() != 0)) {
          oc = FindByProcOrderId(order_info.proc_order_id());
          // handle duplicated proc_order_id by different process
          if (oc && oc->order_spec().product != order_info.symbol()) {
            oc = nullptr;
          }
        }
      }
      if (oc == nullptr) {
        // receive this before submit response
        // exchange does not support client order id
        if (mea().exchange == ExchangeType::Mercado || mea().exchange == ExchangeType::Bithumb) {
          oc = FindSubmitPending(order_info);
        }
      }
    }

    // handle risk pending order
    if (oc) {
      risk_pending_order_->HandleOrderVerified(ts, *oc);
    }

    if (oc != nullptr) {
      if (order_info.is_live()) {
        oc->UpdateLastTimeQueryAlive(ts);
        // If order from query response shows alive but at our side it is moved to dead list
        // already.
        if (live_orders_.find(oc->proc_order_id()) == live_orders_.end()) {
          const double age = (ts - oc->last_seen_time()) / 1e9;
          if (oc->is_alive() && oc->num_of_cancels() <= 3) {
            // oc->is_alive indicates that oc is moved to dead list
            // because 1. treated as zombie order  2. submit order and no accept message is
            // received (include submit respose, query order info, ws, fix).
            if (order_info.has_created_timestamp()) {
              if ((get_ts() - order_info.created_timestamp()) < (3600 * 1e9)) {
                AddBackWorkingOrders(oc);
              }
            } else {
              AddBackWorkingOrders(oc);
            }

            // record abnormal case
            nlohmann::json json;
            json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
            json["error_code"] = "ORDER_CONTEXT_DEAD_ORDER_RESURRECTED";
            result_.AddInfo(LOCATION, json);
          } else if (age > 30 && oc->last_seen_time() > 0) {
            // order_info shows alive, but at our side oc is dead. then we keep track of this
            // event times.
            dead_order_alive_times_map_[oc->proc_order_id()]++;
            SPDLOG_ERROR(
                "Dead order seen in open orders for {} sec, {} times, {}",
                age,
                dead_order_alive_times_map_[oc->proc_order_id()],
                oc->proc_order_id());

            // we need to crash the program in case the dead order is due to fake
            // cancel confirm, thus it will never be canceled unless restarted.
            if (age > dead_order_alive_restart_sec_ &&
                dead_order_alive_times_map_[oc->proc_order_id()] >
                    std::sqrt(dead_order_alive_restart_sec_)) {
              CHECK(false) << "Dead order seen in open orders long enough, Restart!";
            }

            // Record this message.
            nlohmann::json json;
            json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
            json["error_code"] = "ORDER_CONTEXT_DEAD_ORDER_APPEARS_ALIVE";
            result_.AddInfo(LOCATION, json);
          }
        }
      }

      if (order_info.is_live() && oc->status() == OrderSimpleStatus::PENDING_ORDER) {
        // submit response is not received yet, but order status update is received.
        oc->HandleOrderAccepted(ts, order_info.exchange_order_id());
        SetExchangeOrderId(oc, order_info.exchange_order_id());
        // handle cancel accepted for original order of amend order
        HandleAmendOrderAccepted(ts, oc);
      } else if (order_info.cancel_confirmed() && oc->is_alive() && !oc->IsCancelConfirmed()) {
        if (!oc->is_accepted()) {
          // Post-only order, received auto cancel from WS, but rest submit response it not
          // received yet.
          oc->HandleOrderAccepted(ts, order_info.exchange_order_id());
          SetExchangeOrderId(oc, order_info.exchange_order_id());
          // handle cancel accepted for original order of amend order
          HandleAmendOrderAccepted(ts, oc);
        }
        // Cancel not submited, but cancel confirmed received.
        // Must be an auto cancel event.
        if (oc->cancel_manager().status() == CancelStatus::BEGIN) {
          if (order_info.has_canceled_qty()) {
            oc->HandleOrderAutoCanceled(ts, order_info.canceled_qty());
          } else {
            oc->HandleOrderAutoCanceled(ts, oc->working_qty());
          }
        } else {
          if (!oc->IsCancelAccepted()) {
            oc->HandleCancelAccepted(ts);
          }
          oc->HandleCancelConfirmed(ts, oc->working_qty());
        }

        // force set dead
        SetDead(oc->proc_order_id());
        if (!oc->IsClosible()) {
          nlohmann::json json;
          json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
          json["error_code"] = "CANCELED_ORDER_HAS_NONZERO_WORKING_ORDER";
          result_.AddInfo(LOCATION, json);

          SPDLOG_INFO(
              "Cancel confirmed order: {}, working_qty: {}",
              oc->exchange_order_id(),
              oc->working_qty());
        }
      } else if (order_info.fully_filled() && oc->is_alive() && !oc->implied_fill_published()) {
        if (!oc->fill_manager().IsFullyFilled()) {
          if (!oc->is_accepted() && !order_info.exchange_order_id().empty()) {
            SPDLOG_INFO("Fill before accepted! {}", order_info.DebugString());
            oc->HandleOrderAccepted(ts, order_info.exchange_order_id());
            SetExchangeOrderId(oc, order_info.exchange_order_id());
            // handle cancel accepted for original order of amend order
            HandleAmendOrderAccepted(ts, oc);
          }
          std::string fill_id = fmt::format("{}_implied_fill", oc->proc_order_id());
          double filled_qty = oc->order_qty() - oc->filled_qty();
          oc->HandleFill(
              ts,
              oc->order_price(),
              filled_qty,
              fill_id,
              coin::proto::FillType::UNKNOWN_FILL_TYPE,
              coin::proto::FillSource::IMPLIED_FROM_DEAD,
              std::optional<double>{},
              std::optional<std::string>{});
          SPDLOG_INFO("{} {}", fill_id, filled_qty);
        }
        SetDead(oc->proc_order_id());
      }
    } else {
      // treat as abnormal case, should only happen when first time open orders are queried.
      if (order_info.is_live()) {
        auto* new_oc = AddOrderContext(order_info);
        if (order_info.qty() < 1e-10) {
          SPDLOG_ERROR("No qty field in order_info: {}", order_info.DebugString());
        } else {
          SPDLOG_INFO("Unknown order added! {}", order_info.DebugString());
          new_oc->HandleUnknownOrder(ts);
          HandleAmendOrderAccepted(ts, new_oc);
        }
      } else if (mea().exchange == ExchangeType::Deribit) {
        // 20230428(daniel): temporary make can handle foreign orders for Deribit
        if (order_info.has_created_timestamp() &&
            order_info.created_timestamp() > system_start_ts_) {
          auto* new_oc = AddOrderContext(order_info);
          if (order_info.qty() < 1e-10) {
            SPDLOG_ERROR("No qty field in order_info: {}", order_info.DebugString());
          } else {
            SPDLOG_INFO("Unknown order added! {}", order_info.DebugString());
            new_oc->HandleUnknownOrder(ts);
            HandleAmendOrderAccepted(ts, new_oc);
          }
        }
      } else {
        // temporary impl only work for Prex.
        switch (order_info.order_type()) {
          case OrderType::LIQUIDATION_ORDER:
          case OrderType::BLP_TAKEOVER_ORDER: {
            auto* new_oc = AddOrderContext(order_info);
            new_oc->HandleUnknownOrder(ts);
            HandleAmendOrderAccepted(ts, new_oc);
            SetDead(new_oc->proc_order_id());
            nlohmann::json json;
            json["order_context"] = nlohmann::json::parse(ProtoToJsonString(new_oc->ToProto()));
            json["order_info"] = nlohmann::json::parse(ProtoToJsonString(order_info));
            json["error_code"] = "LIQUIDATION_ORDER";
            result_.AddInfo(LOCATION, json);
            break;
          }
          default:
            // ignore over event of unknown dead order first
            break;
        }
      }
    }
  }
}

void OrderContextManager::HandleFillInfoList(const ProductFillBundle& product_fill_bundle) {
  auto now = get_ts();
  auto guard = unknown_fill_id_manager_.GetGuard();
  nlohmann::json json_array = nlohmann::json::array();

  auto order_event_publish_helper = GenerateOrderEventFlusher();

  for (const auto& fill_info : product_fill_bundle.each_fill()) {
    const auto& symbol = fill_info.symbol();
    if (!IsManagedSymbol(symbol)) {
      if (guard.IsNeedLog(symbol, fill_info.fill_id())) {
        nlohmann::json json;
        json["error_code"] = "NOT_MANAGED_SYMBOL";
        json["fill_info"] = nlohmann::json::parse(ProtoToJsonString(fill_info));
        result_.AddInfo(LOCATION, json);
      }
      continue;
    }
    OrderContext* oc = nullptr;
    const auto& exchange_order_id = fill_info.exchange_order_id();
    // TODO(daniel): apply to all exchanges that find order context by proc order id first
    if (mea().exchange == ExchangeType::Prex) {
      oc = FindByOrderIds(
          fill_info.symbol(),
          fill_info.exchange_order_id(),
          fill_info.proc_order_id());
    } else {
      oc = FindByExchangeOrderId(exchange_order_id);
      if (oc == nullptr) {
        if (fill_info.has_proc_order_id() && (fill_info.proc_order_id() != 0)) {
          oc = FindByProcOrderId(fill_info.proc_order_id());
        }
      }
    }
    if (oc == nullptr) {
      if (guard.IsNeedLog(symbol, fill_info.fill_id())) {
        SPDLOG_INFO("Fill msg from unknown order: {}", fill_info.DebugString());
      }
      continue;
    }
    // handle risk pending order
    if (oc) {
      risk_pending_order_->HandleOrderVerified(now, *oc);
    }
    if (oc->order_spec().product != symbol) {
      // Rare cases
      SPDLOG_ERROR(
          "Fill symbol do not match oc: {}\n{}",
          oc->order_spec().DebugString(),
          fill_info.DebugString());

      // should never happen!
      nlohmann::json each_json;
      each_json["symbol"] = symbol;
      each_json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
      json_array.push_back(each_json);
      continue;
    }
    // if ioc fill delay after submit more than 10sec, ignore
    if ((oc->order_spec().duration == coin::proto::OrderDuration::IOC_ORDER ||
         oc->order_spec().duration == coin::proto::OrderDuration::FOK_ORDER) &&
        !oc->has_fill_id(fill_info.fill_id())) {
      auto ts = oc->order_spec().order_created_time;

      if (ts == 0) {
        ts = oc->submit_manager().submitted_time();
      }

      // temperory enlarge ioc/fok fill delay to 30s for prex
      int tolerence_sec = 10;
      if (oc->order_spec().exchange == ExchangeType::Prex) {
        tolerence_sec = 30;
      }
      int age = (now - ts) / 1e9;
      if (age > tolerence_sec) {
        SPDLOG_ERROR("ioc fill delay {} sec, ignore. {}", age, fill_info.DebugString());
        continue;
      }

      nlohmann::json each_json;
      each_json["symbol"] = symbol;
      each_json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
      json_array.push_back(each_json);
    }
    if (!oc->is_accepted() && !exchange_order_id.empty()) {
      SPDLOG_INFO("Fill before accepted! {}", fill_info.DebugString());
      oc->HandleOrderAccepted(now, exchange_order_id);
      SetExchangeOrderId(oc, exchange_order_id);
      // handle cancel accepted for original order of amend order
      HandleAmendOrderAccepted(now, oc);
    }
    std::optional<double> fee;
    if (fill_info.has_fee()) {
      fee = fill_info.fee();
    }
    std::optional<std::string> fee_currency;
    if (fill_info.has_fee_currency()) {
      fee_currency = fill_info.fee_currency();
    }
    oc->HandleFill(
        now,
        fill_info.price(),
        fill_info.qty(),
        fill_info.fill_id(),
        fill_info.fill_type(),
        coin::proto::FillSource::EXPLICIT_FILL_MESSAGE,
        fee,
        fee_currency);
    if (oc->IsClosible()) {
      SetDead(oc->proc_order_id());
    }
  }
}

void OrderContextManager::HandleConversionList(const ConversionBundle& conversion_bundle) {
  auto order_event_publish_helper = GenerateOrderEventFlusher();

  auto now = get_ts();
  for (const auto& conversion : conversion_bundle.each_conversion()) {
    auto conversion_id = conversion.conversion_id();
    if (conversion_set_.count(conversion_id)) {
      continue;
    }
    conversion_set_.emplace(conversion_id);

    auto symbol = fmt::format("{}-{}", conversion.from_currency(), conversion.to_currency());
    coin::proto::OrderEvent oe;
    oe.set_type(OrderEvent::CONVERSION);
    oe.set_external_order_id(conversion_id);
    oe.set_event_time(now);
    oe.set_exchange_type(conversion_bundle.exchange_type());
    oe.set_market_type(conversion_bundle.market_type());
    oe.set_native_symbol(symbol);
    oe.set_symbol(symbol);
    oe.set_fill_price(conversion.exchange_rate());
    oe.set_fill_qty(conversion.from_amount());
    oe.set_fill_id(conversion_id);
    oe.set_fill_type(coin::proto::FillType::TAKER_FILL_TYPE);
    oe.set_fully_filled(true);
    oe.set_publish_live(true);
    oe.set_fill_source(coin::proto::FillSource::EXPLICIT_FILL_MESSAGE);
    PublishOrderEvent(std::move(oe));
  }
}

void OrderContextManager::HandleSubmitResponse(const SubmitOrderResponse& response) {
  auto order_event_publish_helper = GenerateOrderEventFlusher();

  HandleSubmitResponseWithoutFlush(response);
}

void OrderContextManager::HandleSubmitBatchResponse(
    const MultiSubmitResponse& each_submit_response) {
  auto order_event_publish_helper = GenerateOrderEventFlusher();

  for (const auto& submit_response : each_submit_response.each_submit_response()) {
    HandleSubmitResponseWithoutFlush(submit_response);
  }
}

void OrderContextManager::HandleSubmitResponseWithoutFlush(const SubmitOrderResponse& response) {
  /*
   * Must have
   * 1. proc_order_id
   * 2. exchange_order_id
   *
   * */
  CHECK(response.has_proc_order_id()) << response.DebugString();
  CHECK(response.has_timestamp()) << response.DebugString();
  const auto proc_order_id = response.proc_order_id();
  CHECK(HasProcOrderId(proc_order_id)) << response.DebugString();

  const auto& exchange_order_id = response.exchange_order_id();
  auto* oc = FindByProcOrderId(proc_order_id);
  // for handling exchanges that does not support client order id
  auto* oc_alt = FindByExchangeOrderId(exchange_order_id);

  // handle risk pending order
  auto ts = get_ts();
  if (oc) {
    risk_pending_order_->HandleOrderVerified(ts, *oc);
  }
  if (oc_alt) {
    risk_pending_order_->HandleOrderVerified(ts, *oc_alt);
  }

  if (!response.success()) {  // order submission rejected or has some error.
    // handle unsuccessful order submission
    if ((!oc->is_accepted()) && oc->is_alive() &&
        response.error_code() != OrderErrorCode::ORDER_UNCERTAIN) {
      oc->HandleOrderRejected(
          get_ts(),
          response.error_code(),
          response.exchange_error_code(),
          response.exchange_error_msg());
      SetDead(oc->proc_order_id());

      // handle cancel rejected for original order of amend order
      if (oc->IsAmendOrder()) {
        if (auto orig_oc = FindByProcOrderId(oc->order_spec().orig_proc_order_id)) {
          if (orig_oc->is_alive()) {
            orig_oc->HandleCancelRejected(
                get_ts(),
                response.error_code(),
                response.exchange_error_code(),
                response.exchange_error_msg());
          }
        }
      }
    } else {
      // abnormal case
      nlohmann::json json;
      json["error_code"] = "ORDER_CONTEXT_SUBMIT_ORDER_FAILURE";
      json["order_response"] = nlohmann::json::parse(ProtoToJsonString(response));
      result_.AddInfo(LOCATION, json);
    }
    return;
  }

  if (oc->is_accepted()) {
    // This is normal, ws msg often comes earlier.
    SPDLOG_INFO("[OrderContextManager::HandleSubmitResponse] Order is accepted already!");
    return;
  }

  // publish order accepted event
  oc->HandleOrderAccepted(get_ts(), exchange_order_id);
  if (oc_alt && oc_alt != oc && !(oc->order_spec().use_amend)) {
    // Treat as abnormal case, DO NOT UPDATE exchange_order_id_map_!!!
    oc->HandleProcIdReplaced(get_ts(), *oc_alt);
    SetDead(oc->proc_order_id());

    // In some exchange this is normal, let's see how often it happends.
    nlohmann::json json;
    json["error_code"] = "ORDER_CONTEXT_REMOVE_DUPLICATED";
    json["order_response"] = nlohmann::json::parse(ProtoToJsonString(response));
    result_.AddInfo(LOCATION, json);
    return;
  }

  // Update exchange_order_id_map_
  SetExchangeOrderId(oc, exchange_order_id);
  if (oc->order_spec().duration == coin::proto::OrderDuration::IOC_ORDER ||
      oc->order_spec().duration == coin::proto::OrderDuration::FOK_ORDER) {
    oc->HandleOrderAutoCanceled(get_ts(), oc->working_qty());
    SetDead(oc->proc_order_id());

    // how often this happens?
    nlohmann::json json;
    json["error_code"] = "ORDER_CONTEXT_IOC_CREATE_AUTO_CANCEL";
    json["order_response"] = nlohmann::json::parse(ProtoToJsonString(response));
    result_.AddInfo(LOCATION, json);
    return;
  }

  // handle cancel accepted for original order of amend order
  HandleAmendOrderAccepted(get_ts(), oc);
}

void OrderContextManager::HandleCancelResponse(const CancelOrderResponse& response) {
  auto order_event_publish_helper = GenerateOrderEventFlusher();

  HandleCancelResponseWithoutFlush(response);
}

void OrderContextManager::HandleCancelBatchResponse(
    const MultiCancelResponse& each_cancel_response) {
  auto order_event_publish_helper = GenerateOrderEventFlusher();

  for (const auto& cancel_response : each_cancel_response.each_cancel_response()) {
    HandleCancelResponseWithoutFlush(cancel_response);
  }
}

void OrderContextManager::HandleCancelResponseWithoutFlush(const CancelOrderResponse& response) {
  using coin::proto::OrderErrorCode;
  CHECK(response.has_proc_order_id()) << response.DebugString();

  const auto& exchange_order_id = response.exchange_order_id();
  const auto& proc_order_id = response.proc_order_id();

  // TODO(daniel): apply to all exchanges that find order context by proc order id first
  OrderContext* oc = nullptr;
  if (mea().exchange == ExchangeType::Prex) {
    // Daniel(20211018): exchange order id may be shared by different order contexts
    // e.g amend order for prex
    // so changed to find oc by proc order id first (proc order id as unique key of order context)
    if (proc_order_id != 0) {
      oc = FindByProcOrderId(proc_order_id);
    }
    if (oc == nullptr) {
      oc = FindByExchangeOrderId(exchange_order_id);
    }
  } else {
    // Search use exchange order id first, so in case
    // there are two oc corresponds to a single order,
    // (happens if client order id is not supported)
    // we always found the later one.
    oc = FindByExchangeOrderId(exchange_order_id);
    if (oc == nullptr && proc_order_id != 0) {
      oc = FindByProcOrderId(proc_order_id);
    }
  }

  if (!oc) {
    // abnormal, this should never happen.
    SPDLOG_ERROR("[OrderContextManager] Received cancel response of unknown order");

    nlohmann::json json;
    json["error_code"] = "ORDER_CONTEXT_UNKNOWN_ORDER_ID";
    json["cancel_response"] = nlohmann::json::parse(ProtoToJsonString(response));
    // oc == nullptr and crash
    // json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
    result_.AddInfo(LOCATION, json);
    return;
  }

  if (!response.success()) {
    // Cancel reject is normal.
    if (oc->is_alive()) {
      oc->HandleCancelRejected(
          get_ts(),
          response.error_code(),
          response.exchange_error_code(),
          response.exchange_error_msg());
      if (response.is_order_dead()) {
        SetDead(oc->proc_order_id());
      } else if (response.error_code() == OrderErrorCode::ORDER_FILLED_ALREADY) {
        SetDeadWithErrorCode(oc->proc_order_id(), OrderErrorCode::ORDER_FILLED_ALREADY);
      } else if (response.error_code() == OrderErrorCode::ORDER_NOT_FOUND) {
        SetDeadWithErrorCode(oc->proc_order_id(), OrderErrorCode::ORDER_NOT_FOUND);
      } else if (response.error_code() == OrderErrorCode::ORDER_CANCELED_ALREADY) {
        SetDeadWithErrorCode(oc->proc_order_id(), OrderErrorCode::ORDER_CANCELED_ALREADY);
      }
    }
    return;
  }

  auto cancel_status = oc->cancel_manager().status();
  if (cancel_status == CancelStatus::ACCEPTED || cancel_status == CancelStatus::CONFIRMED) {
    // cancel accepted/confirmed is received via ws or rest query
    return;
  }

  if (oc->cancel_manager().status() == CancelStatus::SUBMITTED) {
    if (oc->is_alive()) {
      oc->HandleCancelAccepted(get_ts());
    } else {
      SPDLOG_INFO("Order is dead already! {}", response.DebugString());

      // May indicate that when cancel is on the fly, order is filled.
      // or when last cancel is timed out, we tried to send another cancel,
      // but previous cancel succeeded.
      nlohmann::json json;
      json["error_code"] = "ORDER_CONTEX_ORDER_DEAD_WHEN_CANCEL_SUBMIT_ONFLY";
      json["cancel_response"] = nlohmann::json::parse(ProtoToJsonString(response));
      json["order_context"] = nlohmann::json::parse(ProtoToJsonString(oc->ToProto()));
      result_.AddInfo(LOCATION, json);
    }
  }

  if (response.cancel_confirmed() && !oc->IsCancelConfirmed() && oc->is_alive()) {
    oc->HandleCancelConfirmed(get_ts(), oc->working_qty());
    SetDead(oc->proc_order_id());
  } else {
    // This can be normal
    auto s = fmt::format(
        "only cancel was accepted(not confirmed) or something wrong! response.cancel_confirmed:{}, "
        "oc.state:{}, oc.alive:{}",
        response.cancel_confirmed(),
        int(oc->cancel_manager().status()),
        oc->is_alive());

    /*
    nlohmann::json json;
    json["error_code"] = "CANCEL_ACCEPTED_BUT_NOT_CONFIRMED";
    json["cancel_response"] = nlohmann::json::parse(ProtoToJsonString(response));
    result_.AddInfo(LOCATION, json);
    */
    SPDLOG_INFO(s);
  }
}

void OrderContextManager::HandleBookFeed(const IProduct& product, double ask0, double bid0) {
  auto ts = get_ts();
  for (const auto& it : product_live_orders_.at(product.native_symbol())) {
    auto order = it.second;
    if (order->is_accepted() && !order->IsCancelSent() &&
        order->order_spec().duration != OrderDuration::IOC_ORDER &&
        ts - order->order_accepted_time() > book_feed_fill_interval_ && !order->IsFullyFilled()) {
      bool is_buy = order->order_side() == OrderSide::BUY_ORDER ||
                    order->order_side() == OrderSide::BUY_OPEN_ORDER ||
                    order->order_side() == OrderSide::BUY_CLOSE_ORDER;
      if ((is_buy && bid0 < order->order_price() - 1e-8) ||
          (!is_buy && ask0 > order->order_price() + 1e-8)) {
        SPDLOG_INFO(
            "Implied book fill, ask: {} bid: {} {}",
            ask0,
            bid0,
            order->order_spec().DebugString());
        order->HandleFill(
            ts,
            order->order_price(),
            order->order_qty() - order->filled_qty(),
            fmt::format("{}_implied_book_fill", order->proc_order_id()),
            coin::proto::FillType::MAKER_FILL_TYPE,
            coin::proto::FillSource::IMPLIED_FROM_DEAD,
            std::optional<double>{},
            std::optional<std::string>{});
      }
    }
  }
}

void OrderContextManager::HandleOrderExecution(const OrderExecutionProto& order_execution) {
  auto order_event_publish_helper = GenerateOrderEventFlusher();
  auto ts = get_ts();
  auto err_code = order_execution.error_code().error_code();
  switch (order_execution.type()) {
    case OrderExecutionProto::SUBMIT_ORDER: {
      CHECK(order_execution.has_proc_order_id());
      CHECK_GT(order_execution.proc_order_id(), 0);
      auto order_spec = OrderSpec::FromProto(order_execution.order_spec());
      auto* oc = AddOrderContext(*order_spec);
      oc->HandleOrderSubmitted(ts);
      if (err_code != 0) {
        oc->HandleOrderRejected(
            ts,
            static_cast<coin::proto::OrderErrorCode>(order_execution.error_code().error_code()),
            order_execution.error_code().error_message(),
            order_execution.error_code().category());
        SetDead(oc->proc_order_id());
      }
      break;
    }
    case OrderExecutionProto::CANCEL_ORDER: {
      CHECK(order_execution.has_proc_order_id());
      CHECK_GT(order_execution.proc_order_id(), 0);
      int64_t proc_order_id = order_execution.proc_order_id();
      if (err_code == coin::proto::OrderErrorCode::ORDER_NOT_FOUND) {
        // should never reach here.
        nlohmann::json json;
        json["error_code"] = "ORDER_NOT_FOUND";
        json["order_execution"] = nlohmann::json::parse(ProtoToJsonString(order_execution));
        result_.AddInfo(LOCATION, json);
      } else {
        for (auto poid : order_execution.cancel_timeout_proc_order_ids()) {
          // previous cancel timed-out
          auto* tmp_oc = FindByProcOrderId(poid);
          SetTimeoutCancelRejected(tmp_oc, get_ts());
        }

        auto* oc = FindByProcOrderId(proc_order_id);
        if (oc->is_alive()) {
          oc->HandleCancelSubmitted(ts);
          if (err_code != 0) {
            oc->HandleCancelRejected(
                ts,
                static_cast<coin::proto::OrderErrorCode>(order_execution.error_code().error_code()),
                order_execution.error_code().error_message(),
                order_execution.error_code().category());
          }
        } else {
          // should never reach here.
          nlohmann::json json;
          json["error_code"] = "ORDER_DEAD_ALREADY";
          json["order_execution"] = nlohmann::json::parse(ProtoToJsonString(order_execution));
          result_.AddInfo(LOCATION, json);
        }
      }
      break;
    }
    case OrderExecutionProto::AMEND_ORDER: {
      CHECK(order_execution.has_orig_proc_order_id());
      CHECK_GT(order_execution.orig_proc_order_id(), 0);
      if (err_code == coin::proto::OrderErrorCode::ORDER_NOT_FOUND) {
        // should never reach here.
        nlohmann::json json;
        json["error_code"] = "ORDER_NOT_FOUND";
        json["order_execution"] = nlohmann::json::parse(ProtoToJsonString(order_execution));
        result_.AddInfo(LOCATION, json);
      } else {
        auto amend_order_spec = OrderSpec::FromProto(order_execution.order_spec());
        auto* amend_oc = AddOrderContext(*amend_order_spec);
        amend_oc->HandleOrderSubmitted(ts);
        if (err_code != 0) {
          amend_oc->HandleOrderRejected(
              ts,
              static_cast<coin::proto::OrderErrorCode>(order_execution.error_code().error_code()),
              order_execution.error_code().error_message(),
              order_execution.error_code().category());
          SetDead(amend_oc->proc_order_id());
        } else {
          auto* orig_oc = FindByProcOrderId(order_execution.orig_proc_order_id());
          CHECK(orig_oc);
          orig_oc->HandleCancelSubmitted(ts);
        }
      }
      break;
    }
    case OrderExecutionProto::CANCEL_ALL_ORDERS: {
      for (auto poid : order_execution.cancel_timeout_proc_order_ids()) {
        // previous cancel timed-out
        auto* tmp_oc = FindByProcOrderId(poid);
        SetTimeoutCancelRejected(tmp_oc, get_ts());
      }

      if (err_code > 0) {
        // should never reach here.
        nlohmann::json json;
        json["error_code"] = "ORDER_NOT_FOUND";
        json["order_execution"] = nlohmann::json::parse(ProtoToJsonString(order_execution));
        result_.AddInfo(LOCATION, json);
      } else {
        for (int64_t poid : order_execution.proc_order_ids()) {
          CHECK_GT(poid, 0);
          auto* oc = FindByProcOrderId(poid);
          oc->HandleCancelSubmitted(ts);
        }
      }
      break;
    }
    case OrderExecutionProto::CANCEL_BATCH_ORDERS: {
      for (auto poid : order_execution.cancel_timeout_proc_order_ids()) {
        // previous cancel timed-out
        auto* tmp_oc = FindByProcOrderId(poid);
        SetTimeoutCancelRejected(tmp_oc, get_ts());
      }

      if (err_code == coin::proto::OrderErrorCode::ORDER_NOT_FOUND) {
        // should never reach here.
        nlohmann::json json;
        json["error_code"] = "ORDER_NOT_FOUND";
        json["order_execution"] = nlohmann::json::parse(ProtoToJsonString(order_execution));
        result_.AddInfo(LOCATION, json);
      } else {
        for (const auto& proc_order_id : order_execution.proc_order_ids()) {
          CHECK_GT(proc_order_id, 0);

          auto* oc = FindByProcOrderId(proc_order_id);
          if (oc->is_alive()) {
            oc->HandleCancelSubmitted(ts);
            if (err_code != 0) {
              oc->HandleCancelRejected(
                  ts,
                  static_cast<coin::proto::OrderErrorCode>(
                      order_execution.error_code().error_code()),
                  order_execution.error_code().error_message(),
                  order_execution.error_code().category());
            }
          } else {
            // should never reach here.
            nlohmann::json json;
            json["error_code"] = "ORDER_DEAD_ALREADY";
            json["order_execution"] = nlohmann::json::parse(ProtoToJsonString(order_execution));
            result_.AddInfo(LOCATION, json);
          }
        }
      }
      break;
    }
    case OrderExecutionProto::SUBMIT_BATCH_ORDERS: {
      for (const auto& order_spec_proto : order_execution.order_specs()) {
        auto order_spec = OrderSpec::FromProto(order_spec_proto);

        CHECK_GT(order_spec->proc_order_id, 0);

        auto* oc = AddOrderContext(*order_spec);

        oc->HandleOrderSubmitted(ts);
        if (err_code != 0) {
          oc->HandleOrderRejected(
              ts,
              static_cast<coin::proto::OrderErrorCode>(order_execution.error_code().error_code()),
              order_execution.error_code().error_message(),
              order_execution.error_code().category());
          SetDead(oc->proc_order_id());
        }
      }
      break;
    }
    default: {
      CHECK(false) << order_execution.DebugString();
      break;
    }
  }
}

}  // namespace coin2::exchange::base::order
